package com.sca.bussiness.consumer;

import com.rabbitmq.client.Channel;
import com.sca.bussiness.config.RabbitConfig;
import com.sca.bussiness.entity.AbstractScheduleJob;
import com.sca.bussiness.utils.ApplicationContextHelper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.Serializable;
import java.util.Map;

/**
 * @author liuxianfei
 * @since 2020/11/20 13:54
 */
@Slf4j
@Component
public class ScheduleConsumer {

    /**
     * 定时消息消费
     *
     * @param channel
     * @param headers
     */
    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(value = RabbitConfig.QUEUE_EXEC, durable = "true"),
            exchange = @Exchange(value = RabbitConfig.EXCHG_SCHEDULE, type = "topic"),
            key = RabbitConfig.RTKEY_WAIT_EXEC)
    )
    @RabbitHandler
    public void execConsumer(@Payload Map<String, Serializable> dataMap, Channel channel, @Headers Map<String, Object> headers) {
        long tag = (long) headers.get(AmqpHeaders.DELIVERY_TAG);
        boolean ack = false;
        try {
            log.info("execConsumer接收消息：[{}]", dataMap);
            String triggerClassName = (String) dataMap.get("trigger");
            try {
                @SuppressWarnings("unchecked")
                Class<? extends AbstractScheduleJob> triggerClass = (Class<? extends AbstractScheduleJob>) Class.forName(triggerClassName);
                triggerClass.cast(ApplicationContextHelper.getBean(triggerClass)).executeEntry(dataMap);
            } catch (ClassNotFoundException classNotFoundException) {
                log.error("定时任务业务逻辑类无法找到", classNotFoundException);
            }
            channel.basicAck(tag, false);
            ack = true;
        } catch (IOException ioException) {
            log.error("消息回执出错", ioException);
        } finally {
            if (!ack) {
                try {
                    // 重入队列
                    channel.basicNack(tag, false, true);
                } catch (IOException ioException) {
                    log.error("重入队列异常", ioException);
                }
            }
        }
    }

}
